package com.example.flinktest.transformation;

import com.example.flinktest.source.ClickSource;
import com.example.flinktest.source.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

public class ShuffleTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //读取数据源，并行度为1
        DataStreamSource<Event> streamSource = env.addSource(new ClickSource());
        //随机打印输出，并行度为4
        streamSource.shuffle().print("shuffle").setParallelism(4);

        //1,读取数据源，并行度为1
        DataStreamSource<Event> streamSource1 = env.addSource(new ClickSource());
        //2,轮询重分区后打印输出，并行度为4
        streamSource1.rebalance().print("rebalance").setParallelism(4);
        //3,rescale重缩放分区
        env.addSource(new RichParallelSourceFunction<Integer>() {
            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                for (int i = 0; i < 8; i++) {
                    //将奇数发送到索引为1的并行子任务
                    //将偶数发送到索引为0的并行子任务
                    if ((i + 1) % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                        sourceContext.collect(i + 1);
                    }
                }
            }

            @Override
            public void cancel() {

            }
        }).setParallelism(2).rescale().print().setParallelism(4);
        //4,广播分区
        streamSource.broadcast().print().setParallelism(4);
        //5,全局分区  强行让下游任务并行度变成1，谨慎使用，可能对程序造成很大压力
        streamSource.global().print().setParallelism(4);

        env.execute();
    }
}
